package com.bridgeintelligent.tag.bulkload.service.staticcustomers;

import com.bridgeintelligent.tag.bulkload.api.BulkLoadConfig;
import com.bridgeintelligent.tag.bulkload.api.IndexNameInfo;
import com.bridgeintelligent.tag.bulkload.api.StaticCustomerStatus;
import com.bridgeintelligent.tag.bulkload.api.fileload.FileLoadModel;
import com.bridgeintelligent.tag.bulkload.api.fileload.Filter;
import com.bridgeintelligent.tag.bulkload.api.fileload.FilterGroup;
import com.bridgeintelligent.tag.bulkload.service.core.fileload.ConditionBuilder;
import com.bridgeintelligent.tag.bulkload.service.es.ESUtils;
import com.bridgeintelligent.tag.bulkload.service.es.HiveSqlHelper;
import com.bridgeintelligent.tag.bulkload.service.es.query.EsQueryBuilder;
import com.bridgeintelligent.tag.bulkload.service.looklike.HdfsService;
import com.bridgeintelligent.tag.bulkload.service.looklike.HdfsUtil;
import com.bridgeintelligent.tag.constants.PublicConstant;
import com.bridgeintelligent.tag.constants.bulkload.MQConfig;
import com.bridgeintelligent.tag.constants.bulkload.StaticMqMsg;
import com.bridgeintelligent.tag.mapper.DimDataMapper;
import com.bridgeintelligent.tag.mapper.IssueTaskMapper;
import com.bridgeintelligent.tag.mapper.LookLikeMapper;
import com.bridgeintelligent.tag.mapper.StaticCustomerMapper;
import com.bridgeintelligent.tag.utils.DateHelper;
import com.bridgeintelligent.tag.utils.FieldHandler;
import com.bridgeintelligent.tag.utils.FileUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wayneleo.quickstart.QuickException;
import lombok.extern.slf4j.Slf4j;
import org.apache.lucene.search.TotalHits;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.common.Strings;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermsQueryBuilder;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.*;

import static com.bridgeintelligent.tag.bulkload.api.exception.BulkLoadErrors.*;
import static com.bridgeintelligent.tag.constants.PublicConstant.*;
import static com.bridgeintelligent.tag.utils.DateHelper.PATTERN_DATE_8;

/**
 * @Author：liwei
 * @Date：2023/8/1 14:52
 * @Desc： 静态客群标记处理类
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = MQConfig.STATIC_CUSTOMERS_TOPIC, consumerGroup = MQConfig.STATIC_CUSTOMERS_GROUP,consumeThreadMax=1)
public class StaticCustomersMarkListener implements RocketMQListener<String> {
    private ObjectMapper objectMapper;
    private ConditionBuilder conditionBuilder;
    private LookLikeMapper mapper;
    private StaticCustomerMapper staticCustomerMapper;
    private DimDataMapper dimDataMapper;
    private HiveSqlHelper helper;
    private RocketMQTemplate rocketMQTemplate;
    private HdfsService hdfsService;
    private IssueTaskMapper issueTaskMapper;
    @Value("${nas.remoteStorage:/data}")
    private String remoteStorage;

    public StaticCustomersMarkListener(ObjectMapper objectMapper, ConditionBuilder conditionBuilder, LookLikeMapper mapper, StaticCustomerMapper staticCustomerMapper, DimDataMapper dimDataMapper, HiveSqlHelper helper, RocketMQTemplate rocketMQTemplate, HdfsService hdfsService,IssueTaskMapper issueTaskMapper) {
        this.objectMapper = objectMapper;
        this.conditionBuilder = conditionBuilder;
        this.mapper = mapper;
        this.staticCustomerMapper = staticCustomerMapper;
        this.dimDataMapper = dimDataMapper;
        this.helper = helper;
        this.rocketMQTemplate = rocketMQTemplate;
        this.hdfsService = hdfsService;
        this.issueTaskMapper = issueTaskMapper;
    }

    @Override
    public void onMessage(String s) {
        StaticMqMsg msg = null;
        try {
            msg = objectMapper.readValue(s, StaticMqMsg.class);
            Thread.sleep(1000);
            //幂等性控制，防止单次消费处理时间大于MQ默认15分钟触发重试机制
            int count = issueTaskMapper.transNum(msg.getUniqueIdentify());
            log.info("======>【静态客群ToHdfs】当前线程：{}，处理：{}，获取count:{}",Thread.currentThread().getName(),msg.getUniqueIdentify(),count);
            if (count > 0){
                log.error("任务：{}已经处理过",msg.getUniqueIdentify());
                return;
            }
            issueTaskMapper.addTransNum(msg.getUniqueIdentify());

            if(ONE.equals(msg.getType())){
                log.info("=====>静态客群EsMark接收消息：STATIC_CUSTOMERS_TOPIC:{}>",msg);
                this.doEsMark(msg);
            }else{
                log.info("=====>静态客群HdfsWork接收消息：STATIC_CUSTOMERS_TOPIC:{}>",msg);
                this.doHdfsWork(msg);
            }
        } catch (JsonProcessingException e) {
            log.error("=====>接收静态客群消息[{}]转换失败！！！",msg);
        } catch (Exception e){
            e.printStackTrace();
        }

    }
    private void doEsMark(StaticMqMsg msg){
        String indexId = "";
        IndexNameInfo indexInfo = null;
        try {
            //获取当前客群下的索引信息
            indexId = mapper.findIndexIdByCustomerId(msg.getCustomerId());
            indexInfo = dimDataMapper.findIndexInfoById(indexId);

            staticCustomerMapper.updIsMark(TWO,msg.getCustomerId(),null);//TWO：正在标记
            this.markAndUpdate(msg,indexInfo);//ES查询并标记
            long count = this.getCount(msg,indexInfo.getDataDate(),ONE);//获取标记数量
            log.info("=====>当前静态客群[{}]的标记总数量为：{}",msg.getCustomerId(),count);
            staticCustomerMapper.updIsMark(PublicConstant.ONE,msg.getCustomerId(),count);//ONE：标记成功且更新标记数量
            //this.doWriteHdfs(msg,indexId);//HDFS写文件
        }catch (Exception e){
            log.error("=====>当前静态客群[{}]标记操作失败！！！",msg.getCustomerId());
            try {
                staticCustomerMapper.updIsMark(THREE,msg.getCustomerId(),this.getCount(msg,indexInfo.getDataDate(),ONE));//THREE：标记失败
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
            e.printStackTrace();
        }
    }
    private void doHdfsWork(StaticMqMsg msg){
        String indexId = "";
        IndexNameInfo indexInfo = null;
        try {
            //获取当前客群下的索引信息
            indexId = mapper.findIndexIdByCustomerId(msg.getCustomerId());
            indexInfo = dimDataMapper.findIndexInfoById(indexId);

            this.addStaticCustomer(msg.getCustomerId(),indexId,"",TWO);//正在处理写文件
            long count = getCount(msg,indexInfo.getDataDate(),TWO);
            String hdfsPath = this.hdfsWork(msg,indexInfo.getDataDate(),indexId);//HDFS写文件
            log.info("=====>当前静态客群[{}]写入HDFS文件数据量为：{}",msg.getCustomerId(),count);
            staticCustomerMapper.updStatus(msg.getCustomerId(),ONE,hdfsPath);//ONE：写入成功且更新数量
        } catch (JsonProcessingException e) {
            log.error("=====>接收静态客群消息[{}]转换失败！！！",msg);
        }catch (Exception e){
            log.error("=====>当前静态客群[{}]写HDFS文件操作失败！！！",msg.getCustomerId());
            //THREE：写入失败
            staticCustomerMapper.updStatus(msg.getCustomerId(),THREE,null);//THREE：写入失败
            e.printStackTrace();
        }
    }
    private String hdfsWork(StaticMqMsg msg,String dataDate,String indexId) throws Exception {
        //构建静态客群filterGroup查询条件
        List<Filter> filters = new ArrayList<>();
        FilterGroup filterGroup = conditionBuilder.findFilterGroupAndOverwriteDataDate("-", msg.getCustomerId(), filters).get(0);
        FileLoadModel fileLoad = createFileLoad(EsQueryBuilder.buildFilter(filterGroup));
        //目标文件名
        String fileName = HDFS_STATIC_CUSTOMER_FILENAME_PREFIX+msg.getCustomerId()+"_"+DateHelper.currentDateTime(PATTERN_DATE_8)+"_"+indexId+TXT;
        log.info("=====>静态客群向HDFS写入的文件名：{}",fileName);
//        log.info("=====>[{}]静态客群的HIVESQL为：{}",msg.getCustomerId(),helper.hander(filterGroup));
        //完整的文件路径
        String localPath = remoteStorage+FILESEPERTOR+STATIC_LOCAL_DIR;
        FileUtil.mkdirss(localPath);
        String hdfsPath = hdfsService.getSaticPath(fileName);

        //本地写文件
        this.doWriteLocal(localPath+FILESEPERTOR+fileName,dataDate,fileLoad);
        this.uploadToHdfs(localPath+FILESEPERTOR+fileName,hdfsPath);
        log.info("=====>静态客群向HDFS写入文件的全路径：{}",hdfsPath);
        return hdfsPath;
//        log.info("=====>静态客群向HDFS写入文件的全路径：{}",localPath+FILESEPERTOR+fileName);
//        return localPath+FILESEPERTOR+fileName;
    }
    private FileLoadModel createFileLoad(QueryBuilder queryBuilder) {
        FileLoadModel fileLoadModel = new FileLoadModel();
        fileLoadModel.setQueryBuilder(queryBuilder);
        fileLoadModel.setExcludes(Strings.EMPTY_ARRAY);
        fileLoadModel.setIncludes(new String[]{BulkLoadConfig.CUST_NO});
        return fileLoadModel;
    }
    private void uploadToHdfs(String localPath,String remotePath){
        long startTime = System.currentTimeMillis();
        try {
            HdfsUtil.uploadFile(HdfsUtil.getFileSystem(),localPath,remotePath);
        } catch (Exception e) {
            throw QuickException.normal(ERROR_401004.args(remotePath),e);
        }
        long endTime = System.currentTimeMillis();
        log.info("=====>静态客群上传hdfs文件[{}]成功，耗时：{}s",localPath.substring(localPath.lastIndexOf("/")+1),(endTime - startTime)/1000);
    }
    private void doWriteLocal(String fileName,String dataDate,FileLoadModel fileLoad){
        FileUtil.del(new File(fileName));
        log.info("=====>静态客群本地写文件开始[{}]····",fileName);
        long startTime = System.currentTimeMillis();    //获取ES查询开始时间

        try (BufferedWriter out = new BufferedWriter(
                new FileWriter(fileName, true))){
            ESUtils.scrollSearch(dataDate, fileLoad, hits -> {
                StringBuffer sb = new StringBuffer();
                int i = 0;
                for (SearchHit hit : hits) {
                    Object s = hit.getSourceAsMap().get(BulkLoadConfig.CUST_NO);
                    if (s == null) {
                        continue;
                    }
                    i++;
                    sb.append(s);
                    sb.append("\n");
                    if(i == 5000){
                        out.write(sb.toString());
                        sb.setLength(0);
                    }
                }
                out.write(sb.toString());
                out.flush();
            });
        }catch (IOException e) {
            throw QuickException.normal(ERROR_401005.args(fileName),e);
        }
        long endTime = System.currentTimeMillis();
        log.info("=====>静态客群本地写文件[{}]成功，耗时：{}s",fileName.substring(fileName.lastIndexOf("/")+1),(endTime - startTime)/1000);
    }

    //准备hdfs写文件
    private void doWriteHdfs(StaticMqMsg msg,String indexId){
        //客群历史规则表新增数据--任务开始
        this.addStaticCustomer(msg.getCustomerId(),indexId,"",ZERO);
        try {
            rocketMQTemplate.convertAndSend(MQConfig.STATIC_CUSTOMERS_HDFS_TOPIC, objectMapper.writeValueAsString(msg));
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }catch(Exception e1){
            throw QuickException.normal(ERROR_401007.args(msg.getCustomerId()));
        }
    }
    //查询出静态客群的目标客群并进行标记操作
    private void markAndUpdate(StaticMqMsg msg,IndexNameInfo indexNameInfo){
        //构建静态客群filterGroup查询条件
        List<Filter> filters = new ArrayList<>();
        FilterGroup filterGroup = conditionBuilder.findFilterGroupAndOverwriteDataDate("-", msg.getCustomerId(), filters).get(0);

        //先查询已存在CUSTOMER_FLAG字段的数据集合
        long time = System.currentTimeMillis();
        BoolQueryBuilder queryBuilder = (BoolQueryBuilder)EsQueryBuilder.buildFilter(filterGroup);
        queryBuilder.must(QueryBuilders.existsQuery(FieldHandler.CUSTOMER_FLAG))
        .mustNot(QueryBuilders.termQuery(FieldHandler.CUSTOMER_FLAG,msg.getCustomerId()));
        Script script = new Script("ctx._source['"+FieldHandler.CUSTOMER_FLAG+"'].add('"+msg.getCustomerId()+"');");
        ESUtils.updateByQuery(indexNameInfo.getDataDate(),queryBuilder,script);
        log.info("=====>[{}]静态客群【存在】打标批量耗时：{}s", msg.getCustomerId(),(System.currentTimeMillis() - time)/1000);

        log.info("=====>[{}]静态客群的HIVESQL为：{}",msg.getCustomerId(),helper.hander(filterGroup));
        //再查询并更新不存在CUSTOMER_FLAG字段的数据集合
        long time1 = System.currentTimeMillis();
        BoolQueryBuilder queryBuilder1 = (BoolQueryBuilder)EsQueryBuilder.buildFilter(filterGroup);
        queryBuilder1.mustNot(QueryBuilders.existsQuery(FieldHandler.CUSTOMER_FLAG));
        Script script1 = new Script("ctx._source['"+FieldHandler.CUSTOMER_FLAG+"']=['"+msg.getCustomerId()+"'];");
        ESUtils.updateByQuery(indexNameInfo.getDataDate(),queryBuilder1,script1);
        log.debug("=====>[{}]静态客群【不存在】打标批量耗时：{}s", msg.getCustomerId(),(System.currentTimeMillis() - time1)/1000);
    }

    //客群历史规则表新增数据
    private void addStaticCustomer(String cid,String indexId,String hdfsPath,String status){
        staticCustomerMapper.delCustomerId(cid);
        StaticCustomerStatus entity = new StaticCustomerStatus();
        entity.setCustomerId(cid).setHiveSql(hdfsPath).setIndexId(indexId).setStatus(status).setCreateTime(DateHelper.currentDateTime());
        staticCustomerMapper.add(entity);
    }
    //获取标记数量
    private long getCount(StaticMqMsg msg, String index,String type)throws IOException {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        SearchRequest searchRequest = new SearchRequest(index);
        if(ONE.equals(type)){
            TermsQueryBuilder queryBuilder = QueryBuilders.termsQuery(FieldHandler.CUSTOMER_FLAG,msg.getCustomerId());
            sourceBuilder.query(queryBuilder).trackTotalHits(true).size(0);
        }else{
            List<Filter> filters = new ArrayList<>();
            FilterGroup filterGroup = conditionBuilder.findFilterGroupAndOverwriteDataDate("-", msg.getCustomerId(), filters).get(0);
            sourceBuilder.query(EsQueryBuilder.buildFilter(filterGroup)).trackTotalHits(true).size(0);
        }
        searchRequest.source(sourceBuilder);
        SearchResponse searchResponse = ESUtils.getClient().search(searchRequest, RequestOptions.DEFAULT);
        if (searchResponse != null){
            SearchHits hits = searchResponse.getHits();
            if (hits != null){
                TotalHits totalHits = hits.getTotalHits();
                if (totalHits != null){
                    return totalHits.value;
                }
            }
        }
        return 0l;
    }
}
